package com.mindata.blockchain.socket.client;

import com.google.common.collect.Maps;
import com.mindata.blockchain.common.CommonUtil;
import com.mindata.blockchain.common.ResultPack;
import com.mindata.blockchain.common.SystemConfig;
import com.mindata.blockchain.core.bean.Member;
import com.mindata.blockchain.core.event.NodesConnectedEvent;
import com.mindata.blockchain.core.manager.PermissionManager;
import com.mindata.blockchain.socket.common.Const;
import com.mindata.blockchain.socket.packet.BlockPacket;
import com.mindata.blockchain.socket.packet.NextBlockPacketBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.tio.client.AioClient;
import org.tio.client.ClientGroupContext;
import org.tio.core.Aio;
import org.tio.core.ChannelContext;
import org.tio.core.Node;
import org.tio.utils.lock.SetWithLock;

import javax.annotation.Resource;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.locks.Lock;
import java.util.stream.Collectors;

import static com.mindata.blockchain.socket.common.Const.GROUP_NAME;

/**
 * @author wuweifeng wrote on 2018/3/18.
 */
@Component
public class ClientStarter {
    @Resource
    private ClientGroupContext clientGroupContext;
    @Resource
    private PacketSender packetSender;

    @Resource
    private PermissionManager permissionManager;
    @Resource
    private SystemConfig systemConfig;

    private Logger logger = LoggerFactory.getLogger(getClass());

    private Set<Node> nodes = new HashSet<>();


    // 节点连接状态
    private Map<String, NodeStatus> nodesStatus = Maps.newConcurrentMap();
    private volatile boolean isNodesReady = false; // 节点是否已准备好


    /**
     * 从麦达区块链管理端获取已登记的各服务器ip
     * 隔5分钟去获取一次
     */
    @Scheduled(fixedRate = 10000)
    public void fetchOtherServer() {
        String listenIp = systemConfig.getIp();
        int listenPort = systemConfig.getPort();
        logger.info("本机listenIp：{} listenPort:{}", listenIp, listenPort);

        //获取联盟信息
        ResultPack<List<Member>> resultPack = permissionManager.getMemberDataScheduled();

        //合法的客户端
        if (resultPack.isSucceed()) {
            List<Member> memberList = resultPack.value();
            logger.info("共有" + memberList.size() + "个成员需要连接：" + memberList.toString());

            nodes.clear();
            for (Member member : memberList) {
                Node node = new Node(member.getIp(), member.getPort());
                nodes.add(node);
            }
            //开始尝试绑定到对方开启的server
            bindServerGroup(nodes);

        } else {
            logger.error("不是合法有效的已注册的客户端:{}", resultPack.comment());
            System.exit(0);
        }
    }


    /**
     * 从麦达区块链管理端获取权限信息，一天获取一次即可
     */
    @Scheduled(fixedRate = 1000 * 60 * 60 * 24)
    public void fetchPermission() {
        boolean success = permissionManager.fetchPermission();
        if (!success) {
            logger.error("无法获取权限信息，节点退出");
            System.exit(0);
        }
    }


    /**
     * 每30秒群发一次消息，和别人对比最新的Block
     */
    @Scheduled(fixedRate = 30000)
    public void heartBeat() {
        if (!isNodesReady) return;
        logger.info("---------开始心跳包--------");
        BlockPacket blockPacket = NextBlockPacketBuilder.build();
        packetSender.sendGroup(blockPacket);
    }

    public void onNodesReady() {
        logger.info("开始群发信息获取next Block");
        //在这里发请求，去获取group别人的新区块
        BlockPacket nextBlockPacket = NextBlockPacketBuilder.build();
        packetSender.sendGroup(nextBlockPacket);
    }

    /**
     * client在此绑定多个服务器，多个服务器为一个group，将来发消息时发给一个group。
     * 此处连接的server的ip需要和服务器端保持一致，服务器删了，这边也要踢出Group
     */
    private void bindServerGroup(Set<Node> serverNodes) {
        //当前已经连接的
        SetWithLock<ChannelContext> setWithLock = Aio.getAllChannelContexts(clientGroupContext);
        Lock lock2 = setWithLock.getLock().readLock();
        lock2.lock();
        try {
            Set<ChannelContext> set = setWithLock.getObj();
            //已连接的节点集合
            Set<Node> connectedNodes = set.stream().map(ChannelContext::getServerNode).collect(Collectors.toSet());

            logger.info("已经连接{}个节点:{}", connectedNodes.size(), nodesStatus);

            //连接新增的，删掉已在管理端不存在的
            for (Node node : serverNodes) {
                if (!connectedNodes.contains(node)) {
                    connect(node);
                }
            }
            //删掉服务端已经不存在，或者已经掉线的节点
            for (ChannelContext channelContext : set) {
                Node node = channelContext.getServerNode();
                if (!serverNodes.contains(node)) {
                    Aio.remove(channelContext, "节点不在联盟中主动关闭：" + node);
                }

            }
        } finally {
            lock2.unlock();
        }

    }

    private void connect(Node serverNode) {
        try {
            AioClient aioClient = new AioClient(clientGroupContext);
            logger.info("开始绑定" + ":" + serverNode.toString());
            aioClient.asynConnect(serverNode);
        } catch (Exception e) {
            logger.error("绑定服务[" + serverNode + "]失败", e);
        }
    }

    @EventListener(NodesConnectedEvent.class)
    public void onConnected(NodesConnectedEvent connectedEvent) {
        ChannelContext channelContext = connectedEvent.getSource();
        Node node = channelContext.getServerNode();
        String statusKey = CommonUtil.getNodeStatusKey(node.getIp(), node.getPort());
        if (channelContext.isClosed()) {
            logger.info("连接" + node.toString() + "失败");
            nodesStatus.put(statusKey, NodeStatus.OFF_LINE);
            return;
        } else {
            logger.info("连接" + node.toString() + "成功");
            nodesStatus.put(statusKey, NodeStatus.ON_LINE);
            //绑group是将要连接的各个服务器节点做为一个group
            Aio.bindGroup(channelContext, GROUP_NAME);

            int csize = Aio.getAllChannelContexts(clientGroupContext).size();
            if (csize >= pbftAgreeCount()) {
                synchronized (nodesStatus) {
                    if (!isNodesReady) {
                        isNodesReady = true;
                        onNodesReady();
                    }
                }
            }
        }
    }

    public int halfGroupSize() {
        SetWithLock<ChannelContext> setWithLock = clientGroupContext.groups.clients(clientGroupContext, Const.GROUP_NAME);
        return setWithLock.getObj().size() / 2;
    }

    /**
     * pbft算法中拜占庭节点数量f，总节点数3f+1
     *
     * @return f
     */
    public int pbftSize() {
        //Group内共有多少个节点
        int total = nodes.size();
        int pbft = (total - 1) / 3;
        if (pbft <= 0) {
            pbft = 1;
        }
        //如果要单节点测试，此处返回值改为0
        if (systemConfig.isSingeNode()) {
            return 0;
        }
        return pbft;
    }

    public int pbftAgreeCount() {
        return pbftSize() * 2 + 1;
    }

    /**
     * 节点状态
     */
    enum NodeStatus {
        ON_LINE,
        OFF_LINE;
    }
}
